package ddz.db.redis;

import com.kaka.util.Charsets;
import ddz.utils.ByteUtils;

import java.beans.BeanInfo;
import java.beans.IntrospectionException;
import java.beans.Introspector;
import java.beans.PropertyDescriptor;
import java.lang.reflect.Field;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.Collection;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

public class RedisHashCacher {

    final protected Map<String, PropertyDescriptor> propertyDescriptorMap = new ConcurrentHashMap<>();
    final protected Map<Class<?>, Serializer<?>> serializerMap = new ConcurrentHashMap<>();

    static final class ShortSerializer implements Serializer<Short> {
        @Override
        public byte[] serialize(Short value) {
            if (value == null) return null;
            return ByteUtils.getBytes(value);
        }

        @Override
        public Short deserialize(byte[] bytes) {
            if (bytes == null) return null;
            return ByteUtils.toShort(bytes, 0);
        }
    }

    static final class IntegerSerializer implements Serializer<Integer> {
        @Override
        public byte[] serialize(Integer value) {
            if (value == null) return null;
            return ByteUtils.getBytes(value);
        }

        @Override
        public Integer deserialize(byte[] bytes) {
            if (bytes == null) return null;
            return ByteUtils.toInt(bytes, 0);
        }
    }

    static final class LongSerializer implements Serializer<Long> {
        @Override
        public byte[] serialize(Long value) {
            if (value == null) return null;
            return ByteUtils.getBytes(value);
        }

        @Override
        public Long deserialize(byte[] bytes) {
            if (bytes == null) return null;
            return ByteUtils.toLong(bytes, 0);
        }
    }

    static final class FloatSerializer implements Serializer<Float> {
        @Override
        public byte[] serialize(Float value) {
            if (value == null) return null;
            return ByteUtils.getBytes(value);
        }

        @Override
        public Float deserialize(byte[] bytes) {
            if (bytes == null) return null;
            return ByteUtils.toFloat(bytes, 0);
        }
    }

    static final class DoubleSerializer implements Serializer<Double> {
        @Override
        public byte[] serialize(Double value) {
            if (value == null) return null;
            return ByteUtils.getBytes(value);
        }

        @Override
        public Double deserialize(byte[] bytes) {
            if (bytes == null) return null;
            return ByteUtils.toDouble(bytes, 0);
        }
    }

    static final class BooleanSerializer implements Serializer<Boolean> {
        @Override
        public byte[] serialize(Boolean value) {
            if (value == null) return null;
            return ByteUtils.getBytes(value);
        }

        @Override
        public Boolean deserialize(byte[] bytes) {
            if (bytes == null) return null;
            return ByteUtils.toBoolean(bytes, 0);
        }
    }

    static final class DateSerializer implements Serializer<Date> {
        @Override
        public byte[] serialize(Date value) {
            if (value == null) return null;
            return ByteUtils.getBytes(value.getTime());
        }

        @Override
        public Date deserialize(byte[] bytes) {
            if (bytes == null) return null;
            long millsecs = ByteUtils.toLong(bytes, 0);
            return new Date(millsecs);
        }
    }

    static final class StringSerializer implements Serializer<String> {
        @Override
        public byte[] serialize(String value) {
            if (value == null) return null;
            return value.getBytes(Charsets.utf8);
        }

        @Override
        public String deserialize(byte[] bytes) {
            if (bytes == null) return null;
            return new String(bytes, Charsets.utf8);
        }
    }

    public void init(Class<?> clasz) {
        PropertyDescriptor[] propertyDescriptors = propertyDescriptors(clasz);
        for (PropertyDescriptor pd : propertyDescriptors) {
            propertyDescriptorMap.put(pd.getName(), pd);
            Class<?> propType = pd.getPropertyType();
            Serializer<?> serializer;
            if (propType.equals(Short.TYPE) || propType.equals(Short.class)) {
                serializer = new ShortSerializer();
            } else if (propType.equals(Integer.TYPE) || propType.equals(Integer.class)) {
                serializer = new IntegerSerializer();
            } else if (propType.equals(Long.TYPE) || propType.equals(Long.class)) {
                serializer = new LongSerializer();
            } else if (propType.equals(String.class)) {
                serializer = new StringSerializer();
            } else if (propType.equals(Boolean.TYPE) || propType.equals(Boolean.class)) {
                serializer = new BooleanSerializer();
            } else if (propType.equals(Float.TYPE) || propType.equals(Float.class)) {
                serializer = new FloatSerializer();
            } else if (propType.equals(Double.TYPE) || propType.equals(Double.class)) {
                serializer = new DoubleSerializer();
            } else if (propType.equals(Date.class)) {
                serializer = new DateSerializer();
            } else {
                serializer = new KryoSerializer<>();
            }
            if (!serializerMap.containsKey(propType)) {
                serializerMap.put(pd.getPropertyType(), serializer);
            }
        }
    }

    private PropertyDescriptor[] propertyDescriptors(Class<?> c) throws RuntimeException {
        BeanInfo beanInfo;
        try {
            beanInfo = Introspector.getBeanInfo(c);
        } catch (IntrospectionException e) {
            throw new RuntimeException("Bean introspection failed: " + e.getMessage());
        }
        return beanInfo.getPropertyDescriptors();
    }

    private boolean matchesPrimitive(Class<?> targetType, Class<?> valueType) {
        if (!targetType.isPrimitive()) {
            return false;
        }
        try {
            Field typeField = valueType.getField("TYPE");
            Object primitiveValueType = typeField.get(valueType);
            if (targetType == primitiveValueType) {
                return true;
            }
        } catch (NoSuchFieldException | IllegalAccessException e) {
        }
        return false;
    }

    private boolean isCompatibleType(Object value, Class<?> type) {
        if (value == null || type.isInstance(value) || matchesPrimitive(type, value.getClass())) {
            return true;
        }
        return false;
    }

    final protected void callSetter(Object target, PropertyDescriptor prop, Object value) {
        Method setter = prop.getWriteMethod();
        if (setter == null || setter.getParameterTypes().length != 1) {
            return;
        }
        Class<?> firstParam = setter.getParameterTypes()[0];
        if (this.isCompatibleType(value, firstParam)) {
            try {
                if (firstParam.isArray()) {
                    setter.invoke(target, new Object[]{value});
                } else {
                    setter.invoke(target, value);
                }
            } catch (IllegalAccessException | InvocationTargetException e) {
                throw new RuntimeException(
                        "Cannot set " + prop.getName() + ": " + e.getMessage());
            }
        } else {
            throw new ClassCastException(
                    "Cannot set " + prop.getName() + ": incompatible types, cannot convert "
                            + value.getClass().getName() + " to " + firstParam.getName());
        }
    }

    public void setFieldValue(Object target, String fieldName, byte[] valueBytes) {
        PropertyDescriptor prop = propertyDescriptorMap.get(fieldName);
        Class<?> propCls = prop.getPropertyType();
        Serializer serializer = serializerMap.get(propCls);
        if (serializer == null) return;
        Object value = serializer.deserialize(valueBytes);
        this.callSetter(target, prop, value);
    }

    public Map<byte[], byte[]> toPropBytesMap(Object target) {
        Collection<PropertyDescriptor> pdList = propertyDescriptorMap.values();
        Map<byte[], byte[]> userFieldMap = new HashMap<>(pdList.size());
        for (PropertyDescriptor prop : pdList) {
            Object value;
            try {
                value = prop.getReadMethod().invoke(target);
            } catch (IllegalAccessException | InvocationTargetException e) {
                throw new RuntimeException("反射获取用户字段数据错误", e);
            }
            Class<?> propCls = prop.getPropertyType();
            Serializer serializer = serializerMap.get(propCls);
            if (value != null) {
                byte[] fieldValBytes = serializer.serialize(value);
                if (fieldValBytes != null) {
                    byte[] fieldNameBytes = prop.getName().getBytes(Charsets.utf8);
                    userFieldMap.put(fieldNameBytes, fieldValBytes);
                }
            }
        }
        return userFieldMap;
    }

    /**
     * @param fields 数组元素为name0, value0, name1, value1,……
     * @return
     */
    public Map<byte[], byte[]> toPropBytesMap(Object[] fields) {
        if (fields == null) return null;
        Map<byte[], byte[]> userFieldMap = new HashMap<>(fields.length / 2);
        for (int i = 0; i < fields.length - 1; i += 2) {
            String fieldName = String.valueOf(fields[i]);
            Object fieldVal = fields[i + 1];
            if (fieldVal == null) continue;
            PropertyDescriptor prop = propertyDescriptorMap.get(fieldName);
            if (prop == null) continue;
            Class<?> propCls = prop.getPropertyType();
            Serializer serializer = serializerMap.get(propCls);
            if (serializer == null) continue;
            byte[] valBytes = serializer.serialize(fieldVal);
            if (valBytes == null) continue;
            userFieldMap.put(fieldName.getBytes(Charsets.utf8), valBytes);
        }
        return userFieldMap;
    }

    public Map<byte[], byte[]> toPropBytesMap(Object target, String[] fieldNames) {
        if (fieldNames == null || fieldNames.length == 0) {
            return toPropBytesMap(target);
        }
        Map<byte[], byte[]> userFieldMap = new HashMap<>(fieldNames.length);
        for (int i = 0; i < fieldNames.length; i ++) {
            String fieldName = fieldNames[i];
            PropertyDescriptor prop = propertyDescriptorMap.get(fieldName);
            if (prop == null) continue;
            Object fieldVal;
            try {
                fieldVal = prop.getReadMethod().invoke(target);
            } catch (IllegalAccessException | InvocationTargetException e) {
                continue;
            }
            if (fieldVal == null) continue;
            Class<?> propCls = prop.getPropertyType();
            Serializer serializer = serializerMap.get(propCls);
            if (serializer == null) continue;
            byte[] valBytes = serializer.serialize(fieldVal);
            if (valBytes == null) continue;
            userFieldMap.put(fieldName.getBytes(Charsets.utf8), valBytes);
        }
        return userFieldMap;
    }

}
